{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "wTCNfLLgeOp5"
   },
   "source": [
    "# Install Dependencies\n",
    "\n",
    "If you are running on Google Colab, you need to install the necessary dependencies before beginning the exercise."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "gS2mswUUeMr8"
   },
   "outputs": [],
   "source": [
    "print('NOTE: Intentionally crashing session to use the newly installed library.\\n')\n",
    "\n",
    "!pip uninstall -y pyarrow\n",
    "!pip install ray[debug]==0.7.5\n",
    "\n",
    "# A hack to force the runtime to restart, needed to include the above dependencies.\n",
    "import os\n",
    "os._exit(0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "ei5WPYjRdskC"
   },
   "source": [
    "# Exercise 4 - Introducing Actors\n",
    "\n",
    "**Goal:** The goal of this exercise is to demonstrate how to create stateful actors and call their methods.\n",
    "\n",
    "For more details, please see the documentation on actors: http://ray.readthedocs.io/en/latest/actors.html\n",
    "\n",
    "Although remote functions are useful for parallelizing stateless computations, sometimes your workload requires maintaining state across invocations. Some examples might be a simple counter, a neural network during training, or a simulator environment. If using remote functions, you would have to pass this state into each function invocation and return the updated state when it finishes.\n",
    "\n",
    "However, Ray comes with a stateful abstraction for these situations: remote actors. An actor is a lot like a Python object - it is initialized with an `__init__` function (that has the same features has remote tasks), and can contain internal state that is accessed and mutated by remote method calls. Remote method calls will be executed one at a time on each actor, so there's no need to worry about race conditions on the actor's state. To achieve more parallelism, multiple actors can be created.\n",
    "\n",
    "### Remote Actors\n",
    "\n",
    "To create an actor, decorate a Python class with the `@ray.remote` decorator.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "class Example(object):\n",
    "    def __init__(self, x):\n",
    "        self.x = x\n",
    "    \n",
    "    def set(self, x):\n",
    "        self.x = x\n",
    "    \n",
    "    def get(self):\n",
    "        return self.x\n",
    "```\n",
    "\n",
    "Like regular Python classes, **actors encapsulate state that is shared across actor method invocations**.\n",
    "\n",
    "Actor classes differ from regular Python classes in the following ways.\n",
    "1. **Instantiation:** A regular class would be instantiated via `e = Example(1)`. Actors are instantiated via\n",
    "    ```python\n",
    "    e = Example.remote(1)\n",
    "    ```\n",
    "    When an actor is instantiated, a **new process** is created somewhere in the cluster and the actor __init__ method is run in that process.\n",
    "2. **Method Invocation:** Methods of a regular class would be invoked via `e.set(2)` or `e.get()`. Actor methods are invoked using remote task syntax.\n",
    "    ```python\n",
    "    >>> e.set.remote(2)\n",
    "    ObjectID(d966aa9b6486331dc2257522734a69ff603e5a1c)\n",
    "    \n",
    "    >>> e.get.remote()\n",
    "    ObjectID(7c432c085864ed4c7c18cf112377a608676afbc3)\n",
    "    ```\n",
    "3. **Return Values:** Actor methods are non-blocking. They immediately return an object ID and **create a task that is scheduled on the actor**. The result can be retrieved with `ray.get`.\n",
    "    ```python\n",
    "    >>> ray.get(e.set.remote(2))\n",
    "    None\n",
    "    \n",
    "    >>> ray.get(e.get.remote())\n",
    "    2\n",
    "    ```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "impMwpoqdskF"
   },
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "from collections import defaultdict\n",
    "import numpy as np\n",
    "import time\n",
    "\n",
    "import ray\n",
    "\n",
    "print('Successfully imported ray!')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "qMVSh6bBdskJ"
   },
   "outputs": [],
   "source": [
    "ray.init(num_cpus=4, ignore_reinit_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "B6nnZRgPdskM"
   },
   "source": [
    "**EXERCISE:** Make the `Foo` class an actor class |using the `@ray.remote` decorator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "36mT90AzdskN"
   },
   "outputs": [],
   "source": [
    "class Foo(object):\n",
    "    def __init__(self):\n",
    "        self.counter = 0\n",
    "\n",
    "    def reset(self):\n",
    "        self.counter = 0\n",
    "\n",
    "    def increment(self):\n",
    "        time.sleep(0.5)\n",
    "        self.counter += 1\n",
    "        return self.counter\n",
    "\n",
    "assert hasattr(Foo, 'remote'), 'You need to turn \"Foo\" into an actor with @ray.remote.'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "2x-RmcLHdskQ"
   },
   "source": [
    "**EXERCISE:** Change the intantiations below to create two actors by calling `Foo.remote()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "UEY4DM14dskR"
   },
   "outputs": [],
   "source": [
    "# Create two Foo actors.\n",
    "f1 = Foo()\n",
    "f2 = Foo()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "uNLLwSDedskU"
   },
   "source": [
    "**EXERCISE:** Parallelize the code below. The two actors can execute methods in parallel (though each actor can only execute one method at a time)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "oQ1rPfH-dskU"
   },
   "outputs": [],
   "source": [
    "start_time = time.time()\n",
    "\n",
    "# Reset the actor state so that we can run this cell multiple times without\n",
    "# changing the results.\n",
    "f1.reset()\n",
    "f2.reset()\n",
    "\n",
    "# We want to parallelize this code. However, it is not straightforward to\n",
    "# make \"increment\" a remote function, because state is shared (the value of\n",
    "# \"self.counter\") between subsequent calls to \"increment\". In this case, it\n",
    "# makes sense to use actors.\n",
    "results = []\n",
    "for _ in range(5):\n",
    "    results.append(f1.increment())\n",
    "    results.append(f2.increment())\n",
    "\n",
    "duration = time.time() - start_time\n",
    "assert not any([isinstance(result, ray.ObjectID) for result in results]), 'Looks like \"results\" is {}. You may have forgotten to call ray.get.'.format(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "-NShq-F3dska"
   },
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "w8t7nmbHdskc"
   },
   "outputs": [],
   "source": [
    "assert results == [1, 1, 2, 2, 3, 3, 4, 4, 5, 5]\n",
    "\n",
    "assert duration < 3, ('The experiments ran in {:.3f} seconds. This is too '\n",
    "                      'slow.'.format(duration))\n",
    "assert duration > 2.5, ('The experiments ran in {:.3f} seconds. This is too '\n",
    "                        'fast.'.format(duration))\n",
    "\n",
    "print('Success! The example took {:.3f} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "5PDHBlMRmwlJ"
   },
   "source": [
    "# Exercise 5 - Sharing References to an Actor\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to pass references to actors to remote functions and methods.\n",
    "\n",
    "Sometimes, we may want to have multiple remote tasks that invoke methods on the same actor. For example, we may have a single actor that records logging information for a group of tasks and allows other tasks to query the logs. We can achieve this by passing a handle to the actor (the object returned from calling `Actor.remote()`) as an argument to the tasks.\n",
    "\n",
    "### Actor Handles\n",
    "\n",
    "First, we instantiate an actor:\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "class Actor(object):\n",
    "    def method(self):\n",
    "        pass\n",
    "\n",
    "# Create the actor\n",
    "actor = Actor.remote()\n",
    "```\n",
    "\n",
    "We can define a remote function (or another actor) that takes an actor handle as an argument:\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "def f(actor):\n",
    "    # We can invoke a method on the actor and wait for its result.\n",
    "    return ray.get(actor.method.remote())\n",
    "```\n",
    "\n",
    "This remote function can be invoked multiple times. Each invocation will have a reference to the\n",
    "same actor.\n",
    "\n",
    "```python\n",
    "# Each of the three tasks created below will invoke methods on the same actor.\n",
    "f.remote(actor)\n",
    "f.remote(actor)\n",
    "f.remote(actor)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "gwWAaZU-na-d"
   },
   "source": [
    "In this exercise, we're going to write some code that runs several \"experiments\" in parallel and has each experiment log its results to a shared actor. The main driver script can then periodically pull the results from the logging actor.\n",
    "\n",
    "**EXERCISE:** Turn this `LoggingActor` class into an actor class."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "XEMkg2hTnckg"
   },
   "outputs": [],
   "source": [
    "class LoggingActor(object):\n",
    "    def __init__(self):\n",
    "        self.logs = defaultdict(lambda: [])\n",
    "    \n",
    "    def log(self, index, message):\n",
    "        self.logs[index].append(message)\n",
    "    \n",
    "    def get_logs(self):\n",
    "        return dict(self.logs)\n",
    "\n",
    "\n",
    "assert hasattr(LoggingActor, 'remote'), ('You need to turn LoggingActor into an '\n",
    "                                         'actor (by using the ray.remote keyword).')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "SsuRSVf7nip1"
   },
   "source": [
    "**EXERCISE:** Instantiate the actor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "-edD--YunoXB"
   },
   "outputs": [],
   "source": [
    "logging_actor = LoggingActor()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "OJ08ei6WnprM"
   },
   "source": [
    "Now we define a remote function that runs and pushes its logs to the `LoggingActor`.\n",
    "\n",
    "**EXERCISE:** Modify this function so that it invokes methods correctly on `logging_actor` (you need to change the way you call the `log` method)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "7aV7z23bnt21"
   },
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def run_experiment(experiment_index, logging_actor):\n",
    "    for i in range(60):\n",
    "        time.sleep(1)\n",
    "        # Push a logging message to the actor.\n",
    "        logging_actor.log(experiment_index, 'On iteration {}'.format(i))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "NGv-_hOinxrU"
   },
   "source": [
    "Now we create several tasks that use the logging actor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "vVNhaYfXn49g"
   },
   "outputs": [],
   "source": [
    "experiment_ids = []\n",
    "for i in range(3):\n",
    "    experiment_ids.append(run_experiment.remote(i, logging_actor))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "xzdS-HCJn-iF"
   },
   "source": [
    "While the experiments are running in the background, the driver process (that is, this Jupyter notebook) can query the actor to read the logs.\n",
    "\n",
    "---\n",
    "\n",
    "\n",
    "\n",
    "**EXERCISE:** Modify the code below to fetch logs from the `LoggingActor` using a remote method call."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "KCAjMOZGoBjU"
   },
   "outputs": [],
   "source": [
    "logs = logging_actor.get_logs()\n",
    "\n",
    "assert isinstance(logs, dict), (\"Make sure that you dispatch tasks to the \"\n",
    "                                \"actor using the .remote keyword and get the results using ray.get.\")\n",
    "logs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "zErhr777oFlJ"
   },
   "source": [
    "**EXERCISE:** Try running the above box multiple times and see how the results change (while the experiments are still running in the background). You can also try running more of the experiment tasks and see what happens."
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "Ray Tutorial - Remote Actors",
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "382.391px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
